package com.huitone.smspfm.task;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import com.huitone.smspfm.logger.ExtLogger;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;

import com.bin.common.utils.DateTimeUtils;
import com.bin.common.utils.StringUtils;
import com.huitone.smspfm.core.config.AppStaticConfigs;
import com.huitone.smspfm.core.constant.DealResult;
import com.huitone.smspfm.core.filter.ForbidTimeFilterTask;
import com.huitone.smspfm.core.filter.SmsRouteFilterTask;
import com.huitone.smspfm.model.SmsEntityForMt;
import com.huitone.smspfm.redis.model.AppCanstant;
import com.huitone.smspfm.redis.pipeline.PipelineOpts.ABExcuteCommands;
import com.huitone.smspfm.task.GetPriListLengthTask;

/**   
* @Title: ForbidSmsProcessingManager.java 
* @Package com.huitone.smspfm.module 
* @Description: TODO(用一句话描述该文件做什么) 
* @author ybliang   
* @date 2017年3月15日 下午3:52:00 
* @version V1.0   
*/
public class ForbidProcessingTask extends Thread {

	private static ExtLogger logger = ExtLogger.create(ForbidProcessingTask.class);
	
	private boolean toSleep = false;
	
	private long sleepTime = 1000;
	
	private boolean keepGoing = true;
	
	private static final String CurrFbSpeedS = "CurrFbSpeedS";
	private static final String CurrFbSpeedM = "CurrFbSpeedM";
	
	private static final byte[] CURR_FB_SPEEDS_B = AppStaticConfigs.stringRedisSerializer.serialize(CurrFbSpeedS);
	
	private static final byte[] CURR_FB_SPEEDM_B = AppStaticConfigs.stringRedisSerializer.serialize(CurrFbSpeedM);
	
	private byte[] currFbSpeedS_List = AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.RunningParamsHash + ":" + CurrFbSpeedS);
	private byte[] currFbSpeedM_List = AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.RunningParamsHash + ":" + CurrFbSpeedM);
	
	private long currSpeedS = 0;
	
	private long currSpeedM = 0;
	
	private final Object slock = new Object();
	
	private final Object mlock = new Object();
	
	private ArrayBlockingQueue<Map<byte[],byte[]>> speeds = new ArrayBlockingQueue<Map<byte[],byte[]>>(1000);
	
	private ExecutorService blockingPool = new ThreadPoolExecutor(AppStaticConfigs.mtfbdExcutePoolSize, 
			AppStaticConfigs.mtfbdExcutePoolSize, 1L, TimeUnit.MINUTES, new ArrayBlockingQueue<>(AppStaticConfigs.mtfbdExcutePoolSize * 2), new RejectedExecutionHandler() {
		
		@Override
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			boolean offer = false;
			while(!executor.isShutdown() && !offer) {
				offer = executor.getQueue().offer(r);
				if (!offer) {
					try {
						TimeUnit.MILLISECONDS.sleep(10);
					} catch (InterruptedException e) {}
				}
			}
		}
	});
	
	private ExecutorService blockingPool2 = new ThreadPoolExecutor(AppStaticConfigs.mtfbdExcutePoolSize * 2, 
			AppStaticConfigs.mtfbdExcutePoolSize * 2, 1L, TimeUnit.MINUTES, new ArrayBlockingQueue<>(AppStaticConfigs.mtfbdExcutePoolSize * 4), new RejectedExecutionHandler() {
		
		@Override
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			boolean offer = false;
			while(!executor.isShutdown() && !offer) {
				offer = executor.getQueue().offer(r);
				if (!offer) {
					try {
						TimeUnit.MILLISECONDS.sleep(10);
					} catch (InterruptedException e) {}
				}
			}
		}
	});

	@Override
	public void run() {
		
		if (AppStaticConfigs.speedStatisticsOn) {
			
			startUpdateSpeedTasks();
			
			startSpeedWriteTasks();
			
		}
		
		@SuppressWarnings("rawtypes")
		RedisScript<List> script = new DefaultRedisScript<List>(AppStaticConfigs.luaGetForbid, List.class);
		
		List<String> keys = new ArrayList<>();
		keys.add(AppStaticConfigs.iSMForbidZSet);
		keys.add(AppStaticConfigs.pSMCacheHash);
		
		while(!this.isInterrupted() && isKeepGoing()) {
			
			blockingPool.execute(new Runnable() {
				
				@Override
				public void run() {
					
					try {
						
						Date sdealTime = new Date();
						
						List<List<String>> entities = AppStaticConfigs.scriptOpts.getSMCacheAndHash(script, keys, 0 + "", System.currentTimeMillis() + "", AppStaticConfigs.mtfbdBatchSizePerTime + "");
						
						List<SmsEntityForMt> lists = new ArrayList<>();
						for (int i = 0 ; i < entities.size() ; i ++) {
							List<String> list = entities.get(i);
							SmsEntityForMt smsEntity = SmsEntityForMt.parse(list);
							lists.add(smsEntity);
						}
						
						if (!lists.isEmpty()) {
							
							logger.fbprocc("读取到" + lists.size() + "条不在禁发时段的短信");
							
							int waitCount = 2;
							
							if (AppStaticConfigs.forbidtimeFilterOn) {
								waitCount ++;
							}
							
							CyclicBarrier barrier = new CyclicBarrier(waitCount);
							
							if (AppStaticConfigs.forbidtimeFilterOn) {
								new ForbidTimeFilterTask(lists, barrier).start();
							}
							
							new SmsRouteFilterTask(lists, barrier,false).start();
							
							barrier.await();
							
							Date edealTime = new Date();
							
							blockingPool2.execute(new Runnable() {
								
								@Override
								public void run() {
									
									List<SmsEntityForMt> forbidTimeEntities = new ArrayList<SmsEntityForMt>();
									
									AppStaticConfigs.pipelineOpts3.save(new ABExcuteCommands() {

										@Override
										public void excute(RedisConnection connection) {
											
											Map<String, List<byte[]>> serialnos = new HashMap<String, List<byte[]>>();
											
											for (SmsEntityForMt entity:lists) {
												
												String seqNum = entity.getSerialno() + ":" + entity.getPhoneNumber();
												
												Map<byte[], byte[]> map = entity.getMapB();
												
												map.put(AppCanstant.SmsEntity.SdealTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(sdealTime)));
												map.put(AppCanstant.SmsEntity.EdealTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(edealTime)));
												
												if (entity.isIsforbidsms()) {
													forbidTimeEntities.add(entity);
												} else {
													
													String priListName = entity.getPriListName();
													
													if (!StringUtils.isEmpty(priListName)) {
														// 路由成功
														map.put(AppCanstant.SmsEntity.DealResult, DealResult.RESULT0B);
														map.put(AppCanstant.SmsEntity.DealResultDesc, DealResult.RESULT0_DESCB);
														map.put(AppCanstant.SmsEntity.GateLinkId, AppStaticConfigs.stringRedisSerializer.serialize(entity.getGateLinkId()));
														
														connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + seqNum), map);
														
														if (serialnos.containsKey(priListName)) {
															serialnos.get(priListName).add(AppStaticConfigs.stringRedisSerializer.serialize(seqNum));
														} else {
															List<byte[]> list = new ArrayList<byte[]>();
															list.add(AppStaticConfigs.stringRedisSerializer.serialize(seqNum));
															serialnos.put(priListName, list);
														}
														
														GetPriListLengthTask.priListLength.put(priListName, StringUtils.toLong(GetPriListLengthTask.priListLength.get(priListName)) + 1);
														
														logger.fbprocc("[" + entity.getSerialno() + ":" + entity.getPhoneNumber() + "]读禁发队列路由到" + priListName);
													
													} else {
														// 路由失败
														map.put(AppCanstant.SmsEntity.DealResult, DealResult.RESULT_9B);
														map.put(AppCanstant.SmsEntity.DealResultDesc, DealResult.RESULT_9_DESCB);
														
														connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + seqNum), map);
														
														connection.lPush(AppStaticConfigs.iSMRouteFailedListB, AppStaticConfigs.stringRedisSerializer.serialize(seqNum));
														
														connection.lPush(AppStaticConfigs.iSMNotifyListB, AppStaticConfigs.stringRedisSerializer.serialize(seqNum + ":2"));
													
														logger.fbwarn("[" + entity.getSerialno() + ":" + entity.getPhoneNumber() + "]读禁发队列路由失败.");
													}
												}
											}
											
											// 以批量的方式保存serialno:number到优先级队列
											Iterator<Entry<String, List<byte[]>>> iterator = serialnos.entrySet().iterator();
											while (iterator.hasNext()) {
												
												Entry<String, List<byte[]>> entry = iterator.next();
												
												String key = entry.getKey();
												List<byte[]> value = entry.getValue();
												
												byte[] priListNameB = AppStaticConfigs.stringRedisSerializer.serialize(key);
												
												connection.lPush(priListNameB, value.toArray(new byte[1][]));
												
												connection.sAdd(AppStaticConfigs.iSMPriListNamesSetB, priListNameB);
											}
										}
									});
									
									// 禁发时段短信,保存到禁发队列
									AppStaticConfigs.pipelineOpts.save(new ABExcuteCommands() {

										@Override
										public void excute(RedisConnection connection) {
											
											for (SmsEntityForMt entity:forbidTimeEntities) {
												
												String seqNum = entity.getSerialno() + ":" + entity.getPhoneNumber();
												
												Map<byte[], byte[]> map = entity.getMapB();
												
												map.put(AppCanstant.SmsEntity.SdealTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(sdealTime)));
												map.put(AppCanstant.SmsEntity.EdealTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(edealTime)));
												
												map.put(AppCanstant.SmsEntity.DealResult, DealResult.RESULT_1B);
												map.put(AppCanstant.SmsEntity.DealResultDesc, DealResult.RESULT_1_DESCB);
												
												connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + seqNum), map);
												
												connection.zAdd(AppStaticConfigs.iSMForbidZSetB, entity.getForbidTimeEnd(), AppStaticConfigs.stringRedisSerializer.serialize(seqNum));
												
												logger.mtwarn("[" + entity.getSerialno() + ":" + entity.getPhoneNumber() + "]加入禁发时段队列.");
											}
										}
									});
								}
							});
							
							if (AppStaticConfigs.speedStatisticsOn) {
								int size = lists.size();
								increaseCurrSpeedS(size);
								increaseCurrSpeedM(size);
							}
							
							toSleep = false;
							
						} else {
							toSleep = true;
						}
					} catch (Exception e) {
						logger.error("禁发短信处理过程发生错误:" + e.getMessage());
					}
				}
			});
			
			try {
				if (toSleep) {
					Thread.sleep(sleepTime);
				}
			} catch (InterruptedException e) {
				break;
			}
		}
		
		if (blockingPool != null && !blockingPool.isShutdown()) {
			blockingPool.shutdown();
		}
		
		if (blockingPool2 != null && !blockingPool2.isShutdown()) {
			blockingPool2.shutdown();
		}
		
		logger.info("短信平台禁发队列处理服务已经关闭");
	}
	
	private void startUpdateSpeedTasks() {
		
		new Thread(new Runnable() {
			
			@Override
			public void run() {
				
				while(blockingPool != null && !blockingPool.isShutdown()) {
					
					long speed = getAndResetCurrSpeedS();
					
					speeds.offer(Collections.singletonMap(CURR_FB_SPEEDS_B, AppStaticConfigs.stringRedisSerializer.serialize(speed + ":" + DateTimeUtils.dateToStr2(null))));
					
					logger.debug("上行短信处理速度(条/s):" + speed);
					
					try {
						TimeUnit.MILLISECONDS.sleep(1000);
					} catch (InterruptedException e) {}
				}
			}
		}).start();
		
		new Thread(new Runnable() {
			
			@Override
			public void run() {
				
				int count = 0;
				
				while(blockingPool != null && !blockingPool.isShutdown()) {
					
					if (count >= 60) {
						
						long speed = getAndResetCurrSpeedM();
						
						speeds.offer(Collections.singletonMap(CURR_FB_SPEEDM_B, AppStaticConfigs.stringRedisSerializer.serialize(speed + ":" + DateTimeUtils.dateToStr2(null))));
						
						logger.debug("上行短信处理速度(条/m):" + speed);
						
						count = 0;
						
					} else {
						count ++;
					}
					
					try {
						TimeUnit.MILLISECONDS.sleep(1000);
					} catch (InterruptedException e) {}
				}
			}
		}).start();
	}
	
	private void startSpeedWriteTasks() {
		
		new Thread(new Runnable() {
			
			@Override
			public void run() {
				
				while (blockingPool != null && !blockingPool.isShutdown()) {
					
					Map<byte[],byte[]> map = speeds.poll();
					if (map != null && !map.isEmpty()) {
						AppStaticConfigs.pipelineOpts2.save(new ABExcuteCommands() {

							@Override
							public void excute(RedisConnection connection) {
								connection.hMSet(AppStaticConfigs.RunningParamsHashB, map);
								// 每秒的速率队列
								byte[] bytes = map.get(CURR_FB_SPEEDS_B);
								if (bytes != null) {
									connection.lPush(currFbSpeedS_List, bytes); 
									connection.lTrim(currFbSpeedS_List, 0, AppStaticConfigs.speedSLength - 1);
								}
								// 每分的速率队列
								byte[] bytem = map.get(CURR_FB_SPEEDM_B);
								if (bytem != null) {
									connection.lPush(currFbSpeedM_List, bytem); 
									connection.lTrim(currFbSpeedM_List, 0, AppStaticConfigs.speedMLength - 1);
								}
							}
						});
					} else {
						try {
							TimeUnit.MILLISECONDS.sleep(1000);
						} catch (InterruptedException e) {}
					}
				}
			}
		}).start();
	}

	@Override
	public void destroy() {
		this.setKeepGoing(false);
	}

	public boolean isKeepGoing() {
		return keepGoing;
	}

	public void setKeepGoing(boolean keepGoing) {
		this.keepGoing = keepGoing;
	}
	
	private long getAndResetCurrSpeedS() {
		long result = 0;
		synchronized (slock) {
			result = currSpeedS;
			currSpeedS = 0;
		}
		return result;
	}

	private void increaseCurrSpeedS(long count) {
		synchronized (slock) {
			this.currSpeedS += count;
		}
	}

	public long getCurrSpeedM() {
		return currSpeedM;
	}
	
	private long getAndResetCurrSpeedM() {
		long result = 0;
		synchronized (mlock) {
			result = currSpeedM;
			currSpeedM = 0;
		}
		return result;
	}

	private void increaseCurrSpeedM(long count) {
		synchronized (mlock) {
			this.currSpeedM += count;
		}
	}
}
